package org.yuanzheng.cep

import java.util

import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 10s内连续登录失败3次，判定为恶意登录
 */
case class LoginEvent(id: Long, userName: String, eventType: String, eventTime: Long)

object TestCEP {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    //1定义数据量
    val stream: DataStream[LoginEvent] = streamEnv.fromCollection(List(
      new LoginEvent(1, "张三", "fail", 1577080457),
      new LoginEvent(1, "张三", "fail", 1577080458),
      new LoginEvent(1, "张三", "fail", 1577080460),
      new LoginEvent(1, "李四", "fail", 1577080458),
      new LoginEvent(1, "李四", "success", 1577080462),
      new LoginEvent(1, "张三", "fail", 1577080462)
    )).assignAscendingTimestamps(_.eventTime * 1000L) //指定EventTime的时候必须要确保是时间戳（精确到毫秒）
    //2定义模式
    val pattern: Pattern[LoginEvent, LoginEvent] = Pattern.begin[LoginEvent]("start").where(_.eventType.equals("fail"))
      .next("fail2").where(_.eventType.equals("fail"))
      .next("fail3").where(_.eventType.equals("fail"))
      .within(Time.seconds(10)) //10秒内，基于事件时间
    //3检测模式
    val patternStream: PatternStream[LoginEvent] = CEP.pattern(stream.keyBy(_.userName), pattern) //需要根据用户名分组
    //4选择结果并输出
    val result: DataStream[String] = patternStream.select(new PatternSelectFunction[LoginEvent, String] {
      override def select(map: util.Map[String, util.List[LoginEvent]]): String = {
        val keys: util.Iterator[String] = map.keySet().iterator()
        val e1: LoginEvent = map.get(keys.next()).iterator().next()
        val e2: LoginEvent = map.get(keys.next()).iterator().next()
        val e3: LoginEvent = map.get(keys.next()).iterator().next()
        "用户:" + e1.userName + "，第1次登录失败：" + e1.eventTime + "，第2次登录失败：" + e2.eventTime + "，第3次登录失败：" + e3.eventTime
      }
    })
    result.print()
    streamEnv.execute()
  }
}
